/**
 * TmBatchUpdate.java<br>
 *
 * @author aaronwong.sd@foxmail.com
 * @version 上午10:30:37
 */
package com.iwechat.service.runner;

import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentLinkedQueue;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.iwechat.mybatis.generator.dao.TemplateMessageMapper;
import com.iwechat.mybatis.generator.model.TemplateMessage;

/**
 * 批量更新器
 *
 * @author aaronwong.sd@foxmail.com
 */
@Service
public class TmBatchUpdate implements Runnable {
    private static final Logger                          logger            = LogManager.getLogger();
    private static final long                            WAIT_1S           = 1000;
    private static final long                            INTEVAL_30S       = 30 * WAIT_1S;
    private static final int                             BATCH_COUNT       = 1024;

    @Autowired
    TemplateMessageMapper                                mapper;
    private final ConcurrentLinkedQueue<TemplateMessage> passedMessagePool = MessageSender.passedMessagePool;

    private volatile boolean                             run               = true;

    /*
     * (non-Javadoc)
     *
     * @see java.lang.Runnable#run()
     */
    @Override
    public void run() {
        // 延迟,等待对象初始化
        try {
            synchronized (this) {
                this.wait(WAIT_1S);
            }
        } catch (Exception e) {
        }
        // 循环执行
        long lastUpdateTime = System.currentTimeMillis();
        while (run) {
            try {
                int size = passedMessagePool.size();
                long currentTime = System.currentTimeMillis();
                long inteval = currentTime - lastUpdateTime;
                // 有足够的消息更新
                if (size >= BATCH_COUNT || inteval > INTEVAL_30S) {
                    List<Long> list_status_2 = new LinkedList<Long>();
                    List<Long> list_status_3 = new LinkedList<Long>();

                    TemplateMessage e = null;
                    int k = 0;
                    // passedMessagePool的大小是在变化的,当时我们程序并不会在其他地方对passedMessagePool进行移除操作所以size不会比前值更小
                    while (k < BATCH_COUNT && k < size) {
                        e = passedMessagePool.poll();
                        if (e != null) {
                            if (e.getStatus() == TemplateMessage.Status.STATUS_2_SUCCESS) {
                                list_status_2.add(e.getSeqnb());
                            } else {
                                list_status_3.add(e.getSeqnb());
                            }
                        }
                        k++;
                    }
                    Map<String, Object> map = null;
                    // 成功消息
                    if (list_status_2.size() > 0) {
                        map = new HashMap<String, Object>();
                        map.put("status", TemplateMessage.Status.STATUS_2_SUCCESS);
                        map.put("list", list_status_2);
                        mapper.updateStausBatch(map);
                    }
                    // 失败消息
                    if (list_status_3.size() > 0) {
                        map = new HashMap<String, Object>();
                        map.put("status", TemplateMessage.Status.STATUS_3_FAILED);
                        map.put("list", list_status_3);
                        mapper.updateStausBatch(map);
                    }

                    lastUpdateTime = System.currentTimeMillis();
                    logger.info("batch update [" + list_status_2.size() + "," + list_status_3.size() + "] messages.");
                }
                // 无足够消息
                else {
                    logger.info("Is it this is not enough message to update, and then wait for 30 seconds...");
                    synchronized (this) {
                        this.wait(INTEVAL_30S);
                    }
                }
            } catch (Exception ex) {
                logger.error(ex, ex);
            }
        }

    }

    @PostConstruct
    public void init() {
        new Thread(this, "TempMesgBatchUpdate").start();
        logger.info("### " + this + " init succeed!");
    }

    @PreDestroy
    public void destory() {
        this.run = false;
        synchronized (this) {
            this.notifyAll();
        }
        logger.info("### " + this + " destory succeed!");
    }

}
